package com.niit.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Date:2025/5/21
 * Author：Ys
 * Description:
 */
object SparkStreaming07_reduceByKeyAndWindow {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming07_reduceByKeyAndWindow")
    val ssc = new StreamingContext(conf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")
     //有状态转换 一定要设置检查点，检查点的作用：用来存储（落盘）上次的计算结果数据
    ssc.checkpoint("cp")

    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val wordOne: DStream[(String, Int)] = lines.flatMap(line => line.split(" ")).map(word => (word, 1))
    /*
      reduceByKeyAndWindow:当窗口范围比较大，滑动幅度比较小，可以采用增加数据和删除数据的方法
        没有了重复数据，也不会重复计算
     */
    val resDS: DStream[(String, Int)] = wordOne.reduceByKeyAndWindow(
      (x: Int, y: Int) => {
        x + y
      },
      (x: Int, y: Int) => {
        x - y
      },
      Seconds(9), Seconds(3)
    )
    //resDS.print()
    //foreachRDD 也是输出，常用于将结束数据保存道数据库当中
    resDS.foreachRDD(rdd => {
      rdd.collect().foreach(println)
    })
    ssc.start()

    ssc.awaitTermination()
  }

}
